/**
 * Copyright 2007-2008 Michele Mostarda ( michele.mostarda@gmail.com ).
 * All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the 'License');
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an 'AS IS' BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */


package org.asemantics.anthill;

import org.apache.tools.ant.Project;
import org.jivesoftware.smack.*;
import org.jivesoftware.smack.packet.Message;
import org.jivesoftware.smack.packet.Packet;

import java.net.URI;
import java.net.URISyntaxException;

/**
 * The base class for every Task requiring <i>XMPP</i>
 * connection facilities.
 *
 * @author Michele Mostarda
 */
public abstract class ConnectedTaskBase extends TaskBase {

    /**
     * This class listens for a message response and notifies the waiting <i>thread</i>
     * of the receiving of a response.
     */
    protected class TaskMessageListener implements MessageListener {

        /**
         * The reponse message.
         */
        private Message responseMessage;

        public void processMessage(Chat chat, Message message) {
            synchronized (messageSemaphore) {
                log("received message: " + message.toXML(), Project.MSG_DEBUG);
                responseMessage = message;
                messageSemaphore.notify();
            }
        }

        /**
         * Returns the response message.
         * 
         * @return
         */
        Message getResponseMessage() {
            return responseMessage;
        }

        /**
         * Cleans up the listener data.
         */
        void clear() {
            responseMessage = null;
        }

        /**
         * Returns <code>true</code> if the listener is holding a response message,
         * <code>false</code> otherwise.
         * 
         * @return
         */
        public boolean hasResponse() {
            return responseMessage != null;
        }
    }

    protected class InternalConnectionListener implements ConnectionListener {

        public void connectionClosed() {
            log("connection listener: connection closed", Project.MSG_INFO);
        }

        public void connectionClosedOnError(Exception e) {
            log("connection listener: connection closed on error: " + e.getMessage(), Project.MSG_INFO);
            e.printStackTrace();
            System.exit(1); 
        }

        public void reconnectingIn(int seconds) {
            log("connection listener: reconnecting in " + seconds + " seconds", Project.MSG_INFO);
        }

        public void reconnectionSuccessful() {
            log("connection listener: reconnection successful", Project.MSG_INFO);
        }

        public void reconnectionFailed(Exception e) {
            log("connection listener: reconnection failed: " + e.getMessage(), Project.MSG_INFO);
            throw new RuntimeException(e);
        }
    }

    /**
     * Traces the traffic of every packet received by this task.
     */
    protected class InternalPacketListener implements PacketListener {

        public void processPacket(Packet packet) {
            log("received message from: " + packet.getFrom() );
            log("received packet: " + packet.toXML(), Project.MSG_DEBUG);
        }
    }

    /**
     * Defines the available authorization methods.
     */
    public static enum SecurityType {
        DISABLED {
            public boolean match(String s) {
                return DISABLED_STRING.equalsIgnoreCase(s);
            }
            public String _stringValue() {
                return DISABLED_STRING;
            }
            protected ConnectionConfiguration.SecurityMode getXMPPSecurityMode() {
                return ConnectionConfiguration.SecurityMode.disabled;
            }
        },
        ENABLED{
            public boolean match(String s) {
                return ENABLED_STRING.equalsIgnoreCase(s);
            }
            public String _stringValue() {
                return ENABLED_STRING;
            }
            protected ConnectionConfiguration.SecurityMode getXMPPSecurityMode() {
                return ConnectionConfiguration.SecurityMode.enabled;
            }
        },
        REQUIRED{
            public boolean match(String s) {
                return REQUIRED_STRING.equalsIgnoreCase(s);
            }
            public String _stringValue() {
                return REQUIRED_STRING;
            }
            protected ConnectionConfiguration.SecurityMode getXMPPSecurityMode() {
                return ConnectionConfiguration.SecurityMode.required;
            }
        };

        private static final String DISABLED_STRING = "disabled";
        private static final String ENABLED_STRING  = "enabled";
        private static final String REQUIRED_STRING = "required";

        public abstract boolean match(String s);

        protected abstract String _stringValue();

        protected abstract ConnectionConfiguration.SecurityMode getXMPPSecurityMode();

        public static void validate(String securityType) {
            for(SecurityType type : SecurityType.values()) {
                if(type.match(securityType)) {
                    return;
                }
            }
            throw new IllegalArgumentException("Invalid type: " + securityType);
        }

        public static SecurityType toType(String typeStr) {
            for(SecurityType type : SecurityType.values()) {
                if(type.match(typeStr)) {
                    return type;
                }
            }
            throw new IllegalArgumentException();
        }

        public static String toStringValue(SecurityType st) {
            return st._stringValue();
        }
    }

    /* BEGIN: XMPP constaints. */

    /**
     * The default XMPP server port used if not specified.
     */
    public static final int DEFAULT_XMPP_PORT = 5222;

    /**
     * The default security mode.
     */
    public static final String DEFAULT_SECURITY_MODE = SecurityType.toStringValue(SecurityType.ENABLED);

    /* END: XMPP constats. */

    /* BEGIN: argument constants. */

    /**
     * The default response timeout in milliseconds.
     */
    public static final int RESPONSE_TIMEOUT = 10000;

    /**
     * The number of max tries to execute in sending request message first to raise an exception.
     */
    public static final int MAX_TRIES = 1;

    /* END: argument constants. */


    /**
     * Message semaphore.
     *
     * This object is used to wait for the response message after have sent a request message. 
     */
    protected final Object messageSemaphore = new Object();

    /* BEGIN: node parameters. */

    /**
     * The server URI.
     */
    private URI server;

    /**
     * The server port
     */
    private int port = DEFAULT_XMPP_PORT;

    /**
     * String containing the authorization method type.
     * @see org.asemantics.anthill.ConnectedTaskBase.SecurityType
     */
    private String security = DEFAULT_SECURITY_MODE;

    /**
     * The connection username.
     */
    private String username;

    /**
     * The password username.
     */
    private String password;

    /**
     * Flag indicading wether to enable deugging messages. 
     */
    private boolean debugger = false;

    /**
     * Flag indicating wether to enable message compression.
     */
    private boolean compression = false;

    /* END: node parameters. */

    
    /**
     * The XMPP connection configuration.
     */
    private ConnectionConfiguration configuration;

    /**
     * The XMPP connection.
     */
    private XMPPConnection connection;

    /**
     * The XMPP chat manager.
     */
    private ChatManager chatManager;

    /**
     * Listens to every error occuring on the connection.
     */
    private InternalConnectionListener connectionListener;

    /**
     * The internal packet listener.
     * @see org.asemantics.anthill.ConnectedTaskBase.InternalPacketListener
     */
    private InternalPacketListener packetListener;

    /**
     * Returns the server URL string.
     *
     * @return
     */
    public String getServer() {
        return server.toASCIIString();
    }

    /**
     * Sets the server, validating the URL structure.
     * @param server
     */
    public void setServer(String server) {
        try {
            this.server = new URI(server);
        } catch (URISyntaxException urise) {
            throw new IllegalArgumentException("invalid server value.", urise);
        }
    }

    /**
     * Sets the server port.
     *
     * @param p
     */
    public void setPort(int p) {
        port = p;
    }

    /**
     * Returns the server port.
     *
     * @return
     */
    public int getPort() {
        return port;
    }

    /**
     * Returns the authentication type.
     * 
     * @return
     */
    public String getSecurity() {
        return security;
    }

    /**
     * Sets the authentication type.
     *
     * @param security
     */
    public void setSecurity(String security) {
        SecurityType.validate(security);
        this.security = security;
    }

    /**
     * Returns the username used with the connection.
     *
     * @return
     */
    public String getUsername() {
        return username;
    }

    /**
     * Sets the username to be used with the connetion.
     *
     * @param username
     */
    public void setUsername(String username) {
        if( username.trim().length() == 0 ) {
            throw new IllegalArgumentException("Invalid username: " + username);
        }
        this.username = username;
    }

    /**
     * Returns the password used with the connection.
     *
     * @return
     */
    public String getPassword() {
        return password;
    }

    /**
     * Sets the password used with the connection.
     *
     * @param password
     */
    public void setPassword(String password) {
        this.password = password;
    }

    /**
     * Returns the debugger flag value.
     *
     * @return
     * @see org.asemantics.anthill.ConnectedTaskBase#debugger
     */
    public boolean isDebugger() {
        return debugger;
    }

    /**
     * Sets the debugger glag value.
     *
     * @param debugger
     * @see org.asemantics.anthill.ConnectedTaskBase#debugger
     */
    public void setDebugger(boolean debugger) {
        this.debugger = debugger;
    }

    /**
     * Returns the compression flag value.
     *
     * @return
     * @see org.asemantics.anthill.ConnectedTaskBase#compression
     */
    public boolean isCompression() {
        return compression;
    }

    /**
     * Sets the compression flag value.
     *
     * @param compression
     * @see org.asemantics.anthill.ConnectedTaskBase#compression
     */
    public void setCompression(boolean compression) {
        this.compression = compression;
    }

    /**
     * Creates once and return the <i>XMPP</i> ConnectionConfiguration object.
     *
     * @return
     */
    protected ConnectionConfiguration getConfiguration() {
        if(configuration == null) {
            configuration = new ConnectionConfiguration( getServer(), getPort() );
            configuration.setDebuggerEnabled(debugger);
            configuration.setCompressionEnabled(compression);
            configuration.setSecurityMode( SecurityType.toType(getSecurity()).getXMPPSecurityMode() );
        }
        return configuration;
    }
    
    /**
     * Creates once and return the <i>XMPPConnection</i>.
     *
     * @return
     */
    protected XMPPConnection getConnection() {
        if(connection == null) {
            try {
                connection = new XMPPConnection(getConfiguration());

                log("establishing connection ...", Project.MSG_INFO);
                connection.connect();
                log("connected with server: " + getServer(), Project.MSG_INFO );

                connectionListener = new InternalConnectionListener();
                connection.addConnectionListener(connectionListener);

                packetListener = new InternalPacketListener();
                connection.addPacketListener( packetListener, null );

            } catch (XMPPException xmppe) {
                throw new RuntimeException(
                        "Cannot establish connection with server: " + getServer() + ":" + getPort() + " .", xmppe
                );
            }
        }
        return connection;
    }

    /**
     * Diconnectes the connection if any.
     */
    protected void disconnect() {
        if(connection != null ) {
            if(connection.isConnected()) {
                connection.removeConnectionListener(connectionListener);
                connection.removePacketListener(packetListener);
                connection.disconnect();
                log("connection closed", Project.MSG_INFO);
            }

            connection = null;
        }
    }

    /**
     * performs the login on the connection if needed.
     */
    protected void loginIfNeeded() {
        if(connection == null ) {
            getConnection();
        }
        if( connection.isAuthenticated() ) {
            return;
        }
        try {
            log("logging in ...", Project.MSG_INFO);
            connection.login(getUsername(), getPassword());
            log(" done.", Project.MSG_INFO);
        } catch (XMPPException xmppe) {
            throw new RuntimeException("cannot loginIfNeeded user.");
        }
    }

    /**
     * Creates a message listener and returns it.
     * @return
     */
    protected TaskMessageListener createMessageListener() {
        return new TaskMessageListener();
    }

    /**
     * Creates a Chat and returns it.
     *
     * @param targetUser the user the chat is open for.
     * @return
     */
    protected Chat createChat( String targetUser) {
        loginIfNeeded();
        if(chatManager == null) {
            chatManager = getConnection().getChatManager();
        }
        Chat chat = chatManager.createChat( targetUser, null);
        return chat;
    }

    /**
     * Closes the chat deregistrating the message listener.
     *
     * @param chat
     */
    protected void exitFromChat(Chat chat) {
        //Empty.
    }

    /**
     * Sends a message and wait the current thread until a response message is received
     * or the wait times out.
     *
     * @param chat the chat to use to send the request message
     * @param message the request message
     * @param responseTimeout the response timeout
     * @param maxTries the number of tries to do in seding request message.
     * @return the response message
     */
    protected Message sendMessageAndWaitForReply(
            Chat chat,
            final Message message,
            final int responseTimeout,
            final int maxTries
    ) {
        TaskMessageListener responseListener = createMessageListener();
        chat.addMessageListener(responseListener);
        synchronized (messageSemaphore) {

            // Sending message.
            int retry = 0;
            while(true) {
                if(retry > maxTries) {
                    throw new RuntimeException("Max tries reached during trasmission of message: " + message);
                }
                try {
                    log("sending message:" + message.getBody(), Project.MSG_DEBUG);
                    chat.sendMessage(message);
                    break;
                } catch (Throwable t) {
                    retry++;
                    log("exception " + t.getMessage() + " at try " + retry, Project.MSG_DEBUG);
                    if(isDebugger()) {
                        t.printStackTrace();
                    }
                }
            }

            // Waiting response.
            while (true) {
                try {
                    messageSemaphore.wait(responseTimeout);
                    if( responseListener.hasResponse() ) {
                        break;
                    }
                    throw new RuntimeException("Timeout while waiting reply of message: " + message);
                } catch (InterruptedException ie) {
                    ie.printStackTrace();
                    break;
                } finally {
                    chat.removeMessageListener(responseListener);
                }
            }
            Message response = responseListener.getResponseMessage();
            responseListener.clear();
            return response; 
        }
    }

    /**
     * Sends a message and wait the current thread until a response message is received
     * or the wait times out.
     *
     * @param chat the chat to use to send the request message
     * @param message the request message
     * @return the response message
     */
    protected Message sendMessageAndWaitForReply(Chat chat, Message message ) {
        return sendMessageAndWaitForReply(chat, message, RESPONSE_TIMEOUT, MAX_TRIES);
    }

    /**
     *  Waits for a message on a given chat, returning it as soon as arrives.
     *
     * @param chat the chat to wait for.
     * @return
     */
    protected Message waitForMessage(Chat chat) {
        TaskMessageListener messageListener = createMessageListener();
        chat.addMessageListener(messageListener);
        synchronized (messageSemaphore) {
            while (true) {
                try {
                    messageSemaphore.wait();
                    break;
                } catch (InterruptedException ie) {
                    ie.printStackTrace();
                    break;
                } finally {
                    chat.removeMessageListener(messageListener);
                }
            }
            Message message = messageListener.getResponseMessage();
            messageListener.clear();
            return message;
        }
    }

    /**
     * Unlocks any thread waiting for the semaphore. 
     */
    protected void interruptWaiting() {
        synchronized (messageSemaphore) {
            messageSemaphore.notifyAll();
        }
    }

}
